package com.googlecode.openrtp.grizzly.session;

import com.google.inject.assistedinject.Assisted;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicLong;
import javax.inject.Inject;
import org.glassfish.grizzly.Connection;
import org.glassfish.grizzly.GrizzlyFuture;
import org.glassfish.grizzly.nio.transport.UDPNIOTransport;
import com.googlecode.openrtp.spi.Key;
import com.googlecode.openrtp.spi.RtpSession;
import com.googlecode.openrtp.spi.SessionKeys;
import com.googlecode.openrtp.spi.annotation.Rtcp;
import com.googlecode.openrtp.spi.annotation.Rtp;
import java.util.Objects;
import java.util.logging.Logger;
import org.glassfish.grizzly.Buffer;
import org.glassfish.grizzly.memory.CompositeBuffer;
import org.glassfish.grizzly.memory.MemoryManager;

/**
 *
 * @author jocke
 */
public final class DefaultRtpSession implements RtpSession, PacketGenerator {

    private final ConcurrentHashMap<Key, Object> keys;

    private final UDPNIOTransport rtpTransport;

    private final UDPNIOTransport rtcpTransport;

    private Connection rtcpConnection;

    private Connection rtpConnection;

    private final CompositeBuffer outBuffer;

    private static final Logger LOG = Logger.getLogger(DefaultRtpSession.class.getName());

    @Inject
    public DefaultRtpSession(
            @Rtp UDPNIOTransport rtpTransport,
            @Rtcp UDPNIOTransport rtcpTransport,
            @Assisted("ssrc") Long ssrc) {
        this.keys = new ConcurrentHashMap<>();
        this.rtpTransport = rtpTransport;
        this.rtcpTransport = rtcpTransport;
        set(SessionKeys.RTCP_CONNECTED, Boolean.FALSE);
        set(SessionKeys.RTP_CONNECTED, Boolean.FALSE);
        set(SessionKeys.SEQUNCE_NUMBER, new AtomicLong());
        set(SessionKeys.TOTAL_NUMBER_RTCP_PACKETS, new AtomicLong());
        set(SessionKeys.TOTAL_NUMBER_RTP_PACKETS, new AtomicLong());
        set(SessionKeys.TOTAL_RTP_PACKETS_IN_BYTES, Long.valueOf(0));
        set(SessionKeys.TOTAL_SIZE_RTCP_PACKETS, Long.valueOf(0));
        set(SessionKeys.AVG_SIZE_RTCP, Long.valueOf(0));
        set(SessionKeys.RTP_PACKET_RATE, Long.valueOf(20));
        set(SessionKeys.SSRC, ssrc);
        this.outBuffer = CompositeBuffer.newBuffer();
    }

    @Override
    public void connectRtcp(String host, int port) throws IOException {
        if (!lookup(SessionKeys.RTCP_CONNECTED)) {
            rtcpTransport.start();
            GrizzlyFuture<Connection> connect = rtcpTransport.connect(host, port);
            try {
                this.rtcpConnection = connect.get();
            } catch (InterruptedException | ExecutionException ex) {
                throw new IOException(ex);
            }
        }
    }

    @Override
    public void connectRtp(String host, int port) throws IOException {
        if (!lookup(SessionKeys.RTP_CONNECTED)) {
            set(SessionKeys.RTP_CONNECTED, Boolean.TRUE);
            rtpTransport.start();
            GrizzlyFuture<Connection> connect = rtpTransport.connect(host, port);
            try {
                this.rtpConnection = connect.get();
            } catch (InterruptedException | ExecutionException ex) {
                throw new IOException(ex);
            }
        } else {
            throw new IllegalStateException("already connected");
        }
    }

    @Override
    public void generateRtcpPacket() {
        if (this.rtcpConnection == null) {
            throw new IllegalStateException("RTCP is not connected call connectRtcp(String host, int port) first");
        }
    }

    @Override
    public void generateRtpPacket() {

        if (!outBuffer.hasRemaining()) {
            LOG.fine("Playout buffer empty");

        } else if (this.rtpConnection == null) {
            throw new IllegalStateException("RTP is not connected call connectRtp(String host, int port) first");

        } else {
            rtpConnection.write(this);
        }

    }

    @Override
    public int hashCode() {
        int hash = 3;
        hash = 11 * hash + Objects.hashCode(this.keys);
        hash = 11 * hash + Objects.hashCode(this.rtcpConnection);
        hash = 11 * hash + Objects.hashCode(this.rtpConnection);
        return hash;
    }

    @Override
    public boolean equals(Object obj) {
        if (obj == null) {
            return false;
        }
        if (getClass() != obj.getClass()) {
            return false;
        }
        final DefaultRtpSession other = (DefaultRtpSession) obj;
        if (!Objects.equals(this.keys, other.keys)) {
            return false;
        }
        if (!Objects.equals(this.rtcpConnection, other.rtcpConnection)) {
            return false;
        }
        return Objects.equals(this.rtpConnection, other.rtpConnection);
    }

    @Override
    public <T> T set(Key<T> key, T value) {
        return (T) keys.put(key, value);
    }

    @Override
    public <T> T lookup(Key<T> key) {
        return (T) keys.get(key);
    }

    @Override
    public void appendToOutBuffer(byte[] data) {
        Buffer toAppend = MemoryManager.DEFAULT_MEMORY_MANAGER.allocate(data.length);
        toAppend.put(data);
        toAppend.flip();
        outBuffer.append(toAppend);
    }

    @Override
    public byte[] retrieveDataFromOutBuffer() {
        final int size = Math.min(outBuffer.remaining(), lookup(SessionKeys.RTP_PACKET_SIZE).intValue());
        Buffer split = outBuffer.split(outBuffer.remaining() - size);
        byte[] data = new byte[split.remaining()];
        split.get(data);
        return data;
    }
}
